package cn.com.wind.bdg.inmysql.inmysql.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import cn.com.wind.bdg.inmysql.inmysql.entity.AttachData;
import cn.com.wind.bdg.inmysql.inmysql.entity.HBaseModel;
import cn.com.wind.bdg.inmysql.inmysql.exception.DataInvalidException;
import cn.com.wind.bdg.inmysql.inmysql.exception.DataNotFoundException;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

@Service
public class HBaseServiceImpl implements HBaseService {
    private static Logger LOGGER = LoggerFactory.getLogger(HBaseServiceImpl.class);

    @Autowired
    private Connection hbaseConn;

    public void save(HBaseModel hBaseModel) throws DataInvalidException, IOException {
        String tableNameStr = hBaseModel.getTableName();
        String rowkey = hBaseModel.getRowkey();
        List<AttachData> list = hBaseModel.getAttachDataList();
        if(StringUtils.isBlank(tableNameStr)) {
            throw new DataInvalidException("保存HBase附件时表名为空");
        }
        if(StringUtils.isBlank(rowkey)) {
            throw new DataInvalidException("保存HBase附件时rowkey为空");
        }
        if(CollectionUtils.isEmpty(list)){
            throw new DataInvalidException("保存HBase附件时数据列表为空");
        }
        TableName tableName = TableName.valueOf(tableNameStr);
        Put put = formGet(rowkey,list);
        Table table = hbaseConn.getTable(tableName);
        table.put(put);
    }

    public HBaseModel query(String tableNameStr, String rowkey) throws DataInvalidException,DataNotFoundException,IOException{
        if(StringUtils.isBlank(tableNameStr)) {
            throw new DataInvalidException("保存HBase附件时表名为空");
        }
        if(StringUtils.isBlank(rowkey)) {
            throw new DataInvalidException("保存HBase附件时rowkey为空");
        }
        HBaseModel hBaseModel = new HBaseModel();
        hBaseModel.setRowkey(rowkey);
        hBaseModel.setTableName(rowkey);
        TableName tableName = TableName.valueOf(tableNameStr);
        Table table = hbaseConn.getTable(tableName);
        Get get = new Get(rowkey.getBytes());
        //先判断是否有此条数据
        if(!get.isCheckExistenceOnly()){
            List<AttachData> list = new ArrayList<>();
            Result result = table.get(get);
            Map<String,Map<String,String>> familyMap = new HashMap<>();
            if(result.isEmpty()) {
                throw new DataNotFoundException("Hbase指定rowkey未搜索到，表為:"+tableNameStr+",rowkey為:"+rowkey);
            }
            for (Cell cell : result.rawCells()){
                String family = Bytes.toString(cell.getFamily());
                String colName = Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                if(familyMap.containsKey(family)) {
                    familyMap.get(family).put(colName,value);
                }else {
                    Map<String,String> map = new HashMap<>();
                    map.put(colName,value);
                    familyMap.put(family,map);
                }
            }
            for(Map.Entry<String,Map<String,String>> entry : familyMap.entrySet()) {
                AttachData attachData = new AttachData();
                attachData.setFamilyName(entry.getKey());
                String json = JSON.toJSONString(entry.getValue());
                attachData.setFieldValue(json);
                list.add(attachData);
            }
            hBaseModel.setAttachDataList(list);
        }
        return hBaseModel;
    }

    public void delete(String tableNameStr, String rowkey) throws DataInvalidException,IOException{
        if(StringUtils.isBlank(tableNameStr)) {
            throw new DataInvalidException("保存HBase附件时表名为空");
        }
        if(StringUtils.isBlank(rowkey)) {
            throw new DataInvalidException("保存HBase附件时rowkey为空");
        }
        TableName tableName = TableName.valueOf(tableNameStr);
        Table table = hbaseConn.getTable(tableName);
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        table.delete(delete);
    }


    public Put formGet(String rowkey, List<AttachData> list) {
        Put put = new Put(Bytes.toBytes(rowkey));
        for(AttachData attachData:list){
            Map<String,String> mapData = JSONObject.parseObject(attachData.getFieldValue(),Map.class);
            Iterator iter = mapData.keySet().iterator();
            while (iter.hasNext()) {
                String key = iter.next().toString();
                String value = mapData.get(key);
                put.addColumn(Bytes.toBytes(attachData.getFamilyName()), Bytes.toBytes(key), Bytes.toBytes(value));
            }
        }
//        put.addColumn(Bytes.toBytes("HEAD"), Bytes.toBytes("UP_TIME"), Bytes.toBytes(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss")));
        return put;
    }
}
